package se.sics.kompics.tutorial.tutorial1.appEPFD;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.util.HashSet;
import java.util.Set;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import se.kth.ict.id2203.flp2p.FairLossPointToPointLink;
import se.kth.ict.id2203.flp2p.Flp2pSend;
import se.kth.ict.id2203.pp2p.PerfectPointToPointLink;
import se.kth.ict.id2203.pp2p.Pp2pSend;
import se.sics.kompics.ComponentDefinition;
import se.sics.kompics.Handler;
import se.sics.kompics.Kompics;
import se.sics.kompics.Negative;
import se.sics.kompics.Positive;
import se.sics.kompics.Start;
import se.sics.kompics.address.Address;
import se.sics.kompics.timer.ScheduleTimeout;
import se.sics.kompics.timer.Timer;
import se.sics.kompics.tutorial.basicNetworking.Flp2pMessage;
import se.sics.kompics.tutorial.basicNetworking.Pp2pMessage;
import se.sics.kompics.tutorial.tutorial1.appEPFD.events.RestoredEvent;
import se.sics.kompics.tutorial.tutorial1.appEPFD.events.SuspectedEvent;
import se.sics.kompics.tutorial.tutorial1.appEPFD.messages.HeartbeatLossyMsg;
import se.sics.kompics.tutorial.tutorial1.appEPFD.messages.HeartbeatMessage;
import se.sics.kompics.tutorial.tutorial1.appEPFD.timeout.CheckTimeout;
import se.sics.kompics.tutorial.tutorial1.appEPFD.timeout.HeartbeatTimeout;

public class AppEPFD extends ComponentDefinition {

	Positive<PerfectPointToPointLink> pp2p = requires(PerfectPointToPointLink.class);
	Positive<FairLossPointToPointLink> flp2p = requires(FairLossPointToPointLink.class);
	Positive<Timer> timer = requires(Timer.class);
	Negative<Port_EPFD> EPFD_Port = provides(Port_EPFD.class);
	
	// Delay between two heartbeats
	private static final int TIME_DELAY = 1000;
	
	// Delay between two checks (increased as nodes are wrongly suspected)
	private int period = TIME_DELAY;
	
	// Increase checking period by DELTA in case of misdetection
	private static final int DELTA = 1000;
	
	// Do we use fair loss links or perfect links ?
	private static final boolean LOSSY = true;
	
	private Set<Address> aliveSet; // Heartbeat received since the last check
	private Set<Address> suspectedSet;
	
	private static final Logger logger = LoggerFactory
	.getLogger(AppEPFD.class);
	private String[] commands;
	private int lastCommand;
	private Set<Address> neighborSet;
	private Address self;
	
	public AppEPFD() {
		subscribe(handleInit, control);
		subscribe(handleStart, control);
		subscribe(handleContinue, timer);
		subscribe(handleHeartbeat, timer);
		subscribe(handleCheck, timer);
		subscribe(handleHeartbeatMessage, pp2p);
		subscribe(handleHeartbeatLossyMsg, flp2p);
		//subscribe(handlePp2pMessage, pp2p);
		//subscribe(handleFlp2pMessage, flp2p);
	}
	
	// Init Handler
	Handler<AppEPFD1Init> handleInit = new Handler<AppEPFD1Init>() {
		public void handle(AppEPFD1Init event) {
			commands = event.getCommandScript().split(":");
			lastCommand = -1;
			
			// Topology init
			neighborSet = event.getNeighborSet();
			self = event.getSelf();
			
			// Sets init
			aliveSet = new HashSet<Address>();
			aliveSet.addAll(event.getNeighborSet());
			suspectedSet = new HashSet<Address>();
			
			// Request a heartbeat to the timer
			scheduleHeartbeat();
			
			// Request a check timeout to the timer
			scheduleCheck();
		}
	};
	
	// Heartbeat handler
	Handler<HeartbeatTimeout> handleHeartbeat = new Handler<HeartbeatTimeout>(){

		@Override
		public void handle(HeartbeatTimeout arg0) {
			for(Address address : neighborSet){
				if(!LOSSY){
					// Send Perfect message
					trigger(new Pp2pSend(address, new HeartbeatMessage(self)), pp2p);
				}
				else {
					// Send Lossy message
					trigger(new Flp2pSend(address, new HeartbeatLossyMsg(self)), flp2p);
				}
				
			}
			scheduleHeartbeat();
		}
		
	};
	
	// Check handler
	Handler<CheckTimeout> handleCheck = new Handler<CheckTimeout>(){

		@Override
		public void handle(CheckTimeout arg0) {
			// IF there is nodes both alive and suspected, increase checking period
			HashSet<Address> intersection = new HashSet<Address>(aliveSet);
			intersection.retainAll(suspectedSet);
			if(intersection.size() > 0){
				logger.info("Misdetection occured, increasing period from {} to {}", period, period+DELTA);
				period += DELTA;
			}
			
			for(Address node : neighborSet){
				// If the node is not alive and not yet suspected
				if(!suspectedSet.contains(node) && !aliveSet.contains(node)){
					suspectedSet.add(node);
					logger.info("Neighbor {} is suspected", node);
					trigger(new SuspectedEvent(node), EPFD_Port);
				}
				// If the node is both alive and suspected
				if(suspectedSet.contains(node) && aliveSet.contains(node)){
					suspectedSet.remove(node);
					logger.info("Neighbor {} is not suspected anymore", node);
					trigger(new RestoredEvent(node), EPFD_Port);
				}
			}
			aliveSet.clear();
			scheduleCheck();
		}
		
	};
	
	// Heartbeat reception handler (PP2P)
	Handler<HeartbeatMessage> handleHeartbeatMessage = new Handler<HeartbeatMessage>(){

		@Override
		public void handle(HeartbeatMessage arg0) {
			logger.info("Heartbeat received from {}", arg0.getSource());
			aliveSet.add(arg0.getSource());
		}
	};
	
	// Heartbeat reception handler (FL2P)
	Handler<HeartbeatLossyMsg> handleHeartbeatLossyMsg = new Handler<HeartbeatLossyMsg>(){

		@Override
		public void handle(HeartbeatLossyMsg arg0) {
			logger.info("Heartbeat received from {}", arg0.getSource());
			aliveSet.add(arg0.getSource());
		}
	};
	
	// Start handler
	Handler<Start> handleStart = new Handler<Start>() {
		public void handle(Start event) {
			doNextCommand();
		}
	};
	
	// Continue handler
	Handler<AppEPFDContinue> handleContinue = new Handler<AppEPFDContinue>() {
		public void handle(AppEPFDContinue event) {
			doNextCommand();
		}
	};
	
	private void scheduleHeartbeat() {
		ScheduleTimeout requestHeartbeat = new ScheduleTimeout(TIME_DELAY);
		requestHeartbeat.setTimeoutEvent(new HeartbeatTimeout(requestHeartbeat));
		trigger(requestHeartbeat, timer);
	}

	protected void scheduleCheck() {
		ScheduleTimeout requestCheck = new ScheduleTimeout(period);
		requestCheck.setTimeoutEvent(new CheckTimeout(requestCheck));
		trigger(requestCheck, timer);
	}	
	
	protected void doRecovery(){
		logger.info("I'm back !");
	}
	
	private final void doNextCommand() {
		lastCommand++;

		if (lastCommand > commands.length) {
			return;
		}
		if (lastCommand == commands.length) {
			logger.info("DONE ALL OPERATIONS");
			Thread applicationThread = new Thread("ApplicationThread") {
				public void run() {
					BufferedReader in = new BufferedReader(
							new InputStreamReader(System.in));
					while (true) {
						try {
							String line = in.readLine();
							doCommand(line);
						} catch (Throwable e) {
							e.printStackTrace();
						}
					}
				}
			};
			applicationThread.start();
			return;
		}
		String op = commands[lastCommand];
		doCommand(op);
	}

	private void doCommand(String cmd) {
		if (cmd.startsWith("P")) {
			doPerfect(cmd.substring(1));
			doNextCommand();
		} else if (cmd.startsWith("L")) {
			doLossy(cmd.substring(1));
			doNextCommand();
		} else if (cmd.startsWith("S")) {
			doSleep(Integer.parseInt(cmd.substring(1)));
		} else if (cmd.startsWith("X")) {
			doShutdown();
		} else if (cmd.startsWith("R")) {
			doRecovery();
		}
		
		else if (cmd.equals("help")) {
			doHelp();
			doNextCommand();
		} else {
			logger.info("Bad command: '{}'. Try 'help'", cmd);
			doNextCommand();
		}
	}

	private final void doHelp() {
		logger.info("Available commands: P<m>, L<m>, S<n>, help, X");
		logger.info("Pm: sends perfect message 'm' to all neighbors");
		logger.info("Lm: sends lossy message 'm' to all neighbors");
		logger.info("Sn: sleeps 'n' milliseconds before the next command");
		logger.info("help: shows this help message");
		logger.info("X: terminates this process");
	}

	private final void doPerfect(String message) {
		for (Address neighbor : neighborSet) {
			logger.info("Sending perfect message {} to {}", message, neighbor);
			trigger(new Pp2pSend(neighbor, new Pp2pMessage(self, message)),
					pp2p);
		}
	}

	private final void doLossy(String message) {
		for (Address neighbor : neighborSet) {
			logger.info("Sending lossy message {} to {}", message, neighbor);
			trigger(new Flp2pSend(neighbor, new Flp2pMessage(self, message)),
					flp2p);
		}
	}
	
	private void doSleep(long delay) {
		logger.info("Sleeping {} milliseconds...", delay);

		ScheduleTimeout st = new ScheduleTimeout(delay);
		st.setTimeoutEvent(new AppEPFDContinue(st));
		trigger(st, timer);
	}

	private void doShutdown() {
		System.out.println("2DIE");
		System.out.close();
		System.err.close();
		Kompics.shutdown();
	}
	
}
